package edu.xlh.travel.canal;



import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;;

import java.net.InetSocketAddress;
import java.util.List;

import com.alibaba.otter.canal.protocol.Message;

/**
 * @program: fas-haiot-interface
 * @description: 市级平台数据上传接口相关实现
 * @author: liuAnmin
 * @create: 2021-03-22 15:52
 **/

@Configuration
@Slf4j
public class CanalClient implements ApplicationRunner {

    /*@Resource
    CorpsUploadService corpsUploadService;*/
    @Autowired
    private RedisTemplate redisTemplate;

    @Value("${canal-monitor-mysql.hostname}")
    String canalMonitorHost;

    @Value("${canal-monitor-mysql.port}")
    Integer canalMonitorPort;

    @Value("${canal-monitor-mysql.databasename}")
    String canalMonitorDataBase;

    private final static int BATCH_SIZE = 10000;

    /**
     * 启动服务
     */
    public void startMonitorSQL() {
            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalMonitorHost, canalMonitorPort), "example", "", "");
            try {
                //打开连接
                connector.connect();
                log.info("数据库检测连接成功!" + canalMonitorDataBase);
                //订阅数据库表,全部表q
                connector.subscribe( canalMonitorDataBase+".t_orders");
                //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
                connector.rollback();
                while (true) {
                    // 获取指定数量的数据
                    Message message = connector.getWithoutAck(BATCH_SIZE);
                    long batchId = message.getId();
                    int size = message.getEntries().size();
//                    System.out.println("batchId:"+batchId+"大小:"+size);
                    if (batchId == -1 || size == 0) {
                    } else {
                        handleDATAChange(message.getEntries());
                    }
                    // 提交确认
                    connector.ack(batchId);
                }
            } catch (Exception e) {
                e.printStackTrace();
                log.error("成功断开监测连接!尝试重连");
            } finally {
                connector.disconnect();
                //防止频繁访问数据库链接: 线程睡眠 10秒
                try {
                    log.error("成功断开监测连接!尝试重连");
                    Thread.sleep(10 * 1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
    }


    /**
     * 打印canal server解析binlog获得的实体类信息
     */
    private void handleDATAChange(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }
            //RowChange对象，包含了一行数据变化的所有特征
            CanalEntry.RowChange rowChange;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),e);
            }
            CanalEntry.EventType eventType = rowChange.getEventType();
            log.info("Canal监测到更新:【{}】", entry.getHeader().getTableName());
            switch (eventType) {
                /**
                 * 删除操作
                 */
                case DELETE:
                    System.out.println("canal监控到你删除订单了.....");
                    //corpsUploadService.DeleteOperateToCityInterface(rowChage, entry);
                    break;
                /**
                 * 添加操作
                 */
                case INSERT:
                    System.out.println("canal监控到你下单了.....");
                    //corpsUploadService.InsertOperateToCityInterface(rowChage, entry);
                    break;
                /**
                 * 更新操作
                 */
                case UPDATE:
                    System.out.println("canal监控到你更新订单了.....");
                    handleRedis(rowChange);
                    //corpsUploadService.UpdateOperateToCityInterface(rowChage, entry);
                    break;
                default:
                    break;
            }

        }
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        startMonitorSQL();
    }

    public void handleRedis( CanalEntry.RowChange rowChange) {
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new StringRedisSerializer());
        List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
        for (CanalEntry.RowData rowData : rowDatasList) {
            //一行数据
            List<CanalEntry.Column> oldColumnList  = rowData.getBeforeColumnsList();
            for (CanalEntry.Column column : oldColumnList) {
                if (column.getIsKey()) {
//暂时只支持单一主键
                    Long num = redisTemplate.opsForZSet().remove("orderId", column.getValue());
                    if (num != null && num>0){//解决多线程消费同一个订单
                        System.out.println(Thread.currentThread().getId() + "线程:"+ System.currentTimeMillis()/1000+"ms:redis消费了一个任务：消费的订单OrderId为" +  column.getValue());
                    }
                    break;
                }
            }
        }
    }
}

